﻿using Confluent.Kafka;
using Grow.Binary;
using Grow.Domain.EventBus;
using Grow.EventBus.Cap.PublisherDomainEvents;
using Grow.EventBus.Cap.SubscriberDomainEvents;
using Grow.Module.Service;
using Grow.Utils.Exceptions;
using System;

namespace Grow.EventBus.Cap.Kafka
{
    public class SubscriberDomainEvent<TDomainEventHandler, TDomainEvent>
        : ISubscriberDomainEvent
            where TDomainEventHandler : IDomainEventHandler<TDomainEvent>
            where TDomainEvent : IDomainEvent
    {

        private readonly IServiceContext _serviceContext;
        private readonly IKafkaConnection _kafkaConnection;
        private readonly string _eventName;
        private readonly string _eventHandlerName;
        public SubscriberDomainEvent(IServiceContext serviceContext)
        {
            _eventName = typeof(TDomainEvent).AssemblyQualifiedName!;
            _eventHandlerName = typeof(TDomainEventHandler).AssemblyQualifiedName!;
            _serviceContext = serviceContext;
            _kafkaConnection = _serviceContext.GetRequiredService<IKafkaConnection>();
        }
      
        public async Task  SubscriberAsync(CancellationToken cancellationToken = default)
        {
            cancellationToken.ThrowIfCancellationRequested();
            var eventHandlerOptions = _serviceContext.GetOptions<DomainEventHandlerOptions>(typeof(TDomainEvent).FullName);

            var consumer =  _kafkaConnection.CreateConsumer(eventHandlerOptions.EventHandlerName,
                eventHandlerOptions.EventName);

            consumer.Subscribe(eventHandlerOptions.EventName);

            _ = Task.Run(async () =>
            {
                while (!cancellationToken.IsCancellationRequested)
                {
                    try
                    {
                        await Consume(consumer, eventHandlerOptions.EventName, eventHandlerOptions.EventHandlerName, cancellationToken);
                    }
                    catch (AccessViolationException)
                    {
                        // ignore
                    }
                    catch (Exception)
                    {

                    }
                    // ReSharper disable once MethodSupportsCancellation
                    await Task.Delay(10).ConfigureAwait(false);
                }
            }, cancellationToken);
            await Task.CompletedTask;
        }

        private async Task Consume(IConsumer<string, byte[]> consumer, string eventName, string eventHandlerName,
                    CancellationToken cancellationToken)
        {
            cancellationToken.ThrowIfCancellationRequested();
            ConsumeResult<string, byte[]> consumerResult;
            try
            {
                consumerResult = consumer.Consume(cancellationToken);
                if (consumerResult is null
                    || consumerResult.IsPartitionEOF
                    || consumerResult.Message.Value is null
                    || consumerResult.Message.Value.Length == 0)
                    return;
            }
            catch (OperationCanceledException)
            {
                return;
            }
            catch (Exception ex)
            {
                throw new GrowCheckException($"[EventBus-Kafka] consume message occur error, event: {eventName}, handler: {eventHandlerName}  {ex.Message}");
            }

            try
            {
                var serviceContext = _serviceContext.GetServiceContext();
                var binaryContext= serviceContext.GetRequiredService<IBinaryContext>();
                var domainEventSubscriber = serviceContext.GetRequiredService<IDomainEventSubscriber>();

                var publisherDomainEventDto = await binaryContext.DeserializeAsync<PublisherDomainEventDto>(consumerResult.Message.Value, cancellationToken);
                if (publisherDomainEventDto is null)
                {
                    return;
                }
                var subscriberDomainEventDto = new SubscriberDomainEventDto(
                    publisherDomainEventDto.Id,
                    eventName,
                    _eventName,
                    _eventHandlerName,
                    publisherDomainEventDto.EventBody,
                    SubscriberDomainEventStatus.Wait,
                    0,
                    publisherDomainEventDto.SendAt,
                    null,
                    null
                );
                // 执行消息监听处理
                // 存储偏移,确认消费, see: https://docs.confluent.io/current/clients/dotnet.html
                if (await domainEventSubscriber.SubscriberAsync(subscriberDomainEventDto, cancellationToken))
                {
                    // 只有监听处理成功才提交偏移量,否则不处理即可
                    consumer.StoreOffset(consumerResult);
                }
            }
            catch (Exception ex)
            {
                throw new GrowCheckException($"[EventBus-Kafka] deserialize message from kafka error {ex.Message}");
            }
        }
    }
}